package com.briup.hdfs;

import com.briup.hdfs.pojo.People;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.junit.Test;

import java.io.IOException;

public class SeqStudy {
    @Test
    public void createSequenceFile(){
        // 构建配置
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://192.168.10.11:9000");
        Path output = new Path("/newdata/seq.txt");
        // 指定seq文件要生成到哪
        SequenceFile.Writer.Option seqData =
                SequenceFile.Writer.file(output);
        // 指定键的类型
        SequenceFile.Writer.Option keyClass =
                SequenceFile.Writer.keyClass(IntWritable.class);
        // 指定值的类型
        SequenceFile.Writer.Option valClass =
                SequenceFile.Writer.valueClass(Text.class);
        try {
            // 通过createWriter创建SequenceFile
            SequenceFile.Writer sw =
                    SequenceFile.createWriter(
                            conf,seqData,keyClass,valClass);
            for (int i = 0; i < 100; i++) {
                sw.append(new IntWritable(i),
                        new Text("htl"+i));
            }
            sw.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    @Test
    public void readSequenceFile(){
        // 构建配置
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://192.168.10.11:9000");
        // 指定要读哪个文件
        SequenceFile.Reader.Option seqData =
                SequenceFile.Reader.file(new Path(
                        "/newdata/seq.txt"));
        try {
            // 读取SequenceFile文件
            SequenceFile.Reader sr =
                    new SequenceFile.Reader(conf,seqData);
            // 知道key,value的类型
//            IntWritable keyData = new IntWritable();
//            Text valueData = new Text();
//
//            while(sr.next(keyData,valueData)){
//                System.out.println("key: "+keyData+"value: "+valueData);
//            }
            // 不知道key,value的类型
            // 获取Key和value的类型
            Writable keyClass = (Writable)
                    sr.getKeyClass().newInstance();
            Writable valueClass = (Writable)
                    sr.getValueClass().newInstance();
            while(sr.next(keyClass,valueClass)){
                System.out.println("key: "+keyClass+
                        " value: "+valueClass);
            }

        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (InstantiationException e) {
            throw new RuntimeException(e);
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        }

    }
    @Test
    public void createSeqRecord(){
        // 构建配置
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://192.168.10.11:9000");
        // 指定输出路径
        Path path = new Path("/newdata/block.txt");
        SequenceFile.Writer.Option output =
                SequenceFile.Writer.file(path);
        // 指定key的类型
        SequenceFile.Writer.Option keyClass =
                SequenceFile.Writer.keyClass(
                        IntWritable.class);
        // 指定value的类型
        SequenceFile.Writer.Option valueClass =
                SequenceFile.Writer.valueClass(Text.class);
        // 指定一个record压缩
//        SequenceFile.Writer.Option record = SequenceFile.Writer.compression(
//                SequenceFile.CompressionType.RECORD,
//                new GzipCodec()
//                );
        // 指定块压缩
        SequenceFile.Writer.Option block =
                SequenceFile.Writer.compression(
                SequenceFile.CompressionType.BLOCK,
                new BZip2Codec()
        );
        try {
            SequenceFile.Writer sw =
                    SequenceFile.createWriter(conf,output,
                    keyClass,valueClass,block);

            for (int i = 0; i < 100; i++) {
                sw.append(new IntWritable(i),
                        new Text("xyy"+i));
            }
            sw.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }

    @Test
    public void seqSyncCreate(){
        // 构建配置
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://192.168.10.11:9000");
        Path output = new Path("/newdata/seq.txt");
        // 指定seq文件要生成到哪
        SequenceFile.Writer.Option seqData =
                SequenceFile.Writer.file(output);
        // 指定键的类型
        SequenceFile.Writer.Option keyClass =
                SequenceFile.Writer.keyClass(IntWritable.class);
        // 指定值的类型
        SequenceFile.Writer.Option valClass =
                SequenceFile.Writer.valueClass(Text.class);
        try {
            // 通过createWriter创建SequenceFile
            SequenceFile.Writer sw =
                    SequenceFile.createWriter(
                            conf,seqData,keyClass,valClass);
            for (int i = 0; i < 100; i++) {
                // 创建同步标识
                if (i%2 == 0) sw.sync();
                sw.append(new IntWritable(i),
                        new Text("htl"+i));
            }
            sw.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    @Test
    public void seqSyncRead(){
        // 构建配置
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://192.168.10.11:9000");
        // 指定要读哪个文件
        SequenceFile.Reader.Option seqData =
                SequenceFile.Reader.file(new Path(
                        "/newdata/seq.txt"));
        try {
            // 读取SequenceFile文件
            SequenceFile.Reader sr =
                    new SequenceFile.Reader(conf,seqData);
            // 知道key,value的类型
//            IntWritable keyData = new IntWritable();
//            Text valueData = new Text();
//
//            while(sr.next(keyData,valueData)){
//                System.out.println("key: "+keyData+"value: "+valueData);
//            }
            // 不知道key,value的类型
            // 获取Key和value的类型
            Writable keyClass = (Writable)
                    sr.getKeyClass().newInstance();
            Writable valueClass = (Writable)
                    sr.getValueClass().newInstance();
            // 获取同步标识位置
            long position = sr.getPosition();
            // 跳转
            sr.sync(position);
            while(sr.next(keyClass,valueClass)){
                System.out.println("key: "+keyClass+
                        " value: "+valueClass);
                position = sr.getPosition();
                System.out.println(position);
                sr.sync(position);
            }

        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (InstantiationException e) {
            throw new RuntimeException(e);
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        }

    }
    @Test
    public void peopleSeqWriter(){
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://192.168.10.11:9000");
        Path path = new Path("/newdata/people.txt");
        SequenceFile.Writer.Option output =
                SequenceFile.Writer.file(path);
        SequenceFile.Writer.Option keyClass =
                SequenceFile.Writer.keyClass(IntWritable.class);
        SequenceFile.Writer.Option valClass =
                SequenceFile.Writer.valueClass(People.class);
        try {
            SequenceFile.Writer data =
                    SequenceFile.createWriter(
                            conf,output,keyClass,valClass);
            for (int i = 0; i < 100; i++) {
                People people =
                        new People(i,"葫芦娃",i,"M");
                data.append(new IntWritable(i),people);
            }
            data.close();

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    @Test
    public void readPeopleSequenceFile(){
        // 构建配置
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://192.168.10.11:9000");
        // 指定要读哪个文件
        SequenceFile.Reader.Option seqData =
                SequenceFile.Reader.file(new Path(
                        "/newdata/people.txt"));
        try {
            // 读取SequenceFile文件
            SequenceFile.Reader sr =
                    new SequenceFile.Reader(conf,seqData);
            // 知道key,value的类型
//            IntWritable keyData = new IntWritable();
//            Text valueData = new Text();
//
//            while(sr.next(keyData,valueData)){
//                System.out.println("key: "+keyData+"value: "+valueData);
//            }
            // 不知道key,value的类型
            // 获取Key和value的类型
            Writable keyClass = (Writable)
                    sr.getKeyClass().newInstance();
            Writable valueClass = (Writable)
                    sr.getValueClass().newInstance();
            while(sr.next(keyClass,valueClass)){
                System.out.println("key: "+keyClass+
                        " value: "+valueClass);
            }

        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (InstantiationException e) {
            throw new RuntimeException(e);
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        }

    }


}
